package org.example.双锁实现;

import org.example.BlockingQueue;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BlockingQueue2<E> implements BlockingQueue<E> {

    private E[] array;
    private int head;
    private int tail;
    private AtomicInteger size;

    @SuppressWarnings("all")
    public BlockingQueue2(int capacity) {
        array = (E[]) new Object[capacity];
    }

    private ReentrantLock headLock = new ReentrantLock();
    private Condition headWaits = headLock.newCondition();


    private ReentrantLock tailLock = new ReentrantLock();
    private Condition tailWaits = tailLock.newCondition();

    protected boolean isEmpty() {
        return size.get() == 0;
    }

    protected boolean isFull() {
        return size.get() == array.length;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        try {
            // 1. 队列满则等待
            while (isFull()) {
                tailWaits.await();
            }

            // 2. 队列不满则入队
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }

            // 3. 修改size
            c = size.getAndIncrement();

            if (c + 1 == array.length) {
                tailWaits.signal();
            }
        } finally {
            tailLock.unlock();
        }


        // 4. 唤醒poll 中等待的线程
        if (c == 0) { // 当队列由空转为非空时需 加锁唤醒offer中等待的线程
            headLock.lock();
            try {
                headWaits.signal(); // try finally 快捷键： ctrl+alt+t
            } finally {
                headLock.unlock();
            }
        }
    }

    /**
     *
     * @param e
     * @param timeout 超过多少时间就放弃等待
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        try {
            long t = TimeUnit.MILLISECONDS.toNanos(timeout);

            // 1. 队列满则等待
            while (isFull()) {
                if (t <= 0) {
                    return false;
                }
                t = tailWaits.awaitNanos(t); // 返回的t是剩余等待时间
            }

            // 2. 队列不满则入队
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }

            // 3. 修改size
            c = size.getAndIncrement();


        } finally {
            tailLock.unlock();
        }

        // 4. 唤醒poll 中等待的线程
        if (c == 0) { // 当队列由空转为非空时需 加锁唤醒offer中等待的线程
            headLock.lock();
            try {
                headWaits.signal(); // try finally 快捷键： ctrl+alt+t
            } finally {
                headLock.unlock();
            }
        }

        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        E e = null;
        headLock.lockInterruptibly();
        int c;  // 级联唤醒
        try {
            // 1. 队列空则等待
            while (isEmpty()) {
                headWaits.await();
            }

            // 2. 非空则出队
            e = array[head];
            array[head] = null; // help GC
            if (++head == array.length) {
                head = 0;
            }

            // 3. 修改size
            c = size.getAndDecrement();

            if (c > 1) {
                headWaits.signal();
            }

        } finally {
            headLock.unlock();
        }

        // 4. 唤醒poll 中等待的线程
        if (c == array.length) { // 当队列由满转为不满时，需要唤醒offer中等待的线程
            tailLock.lock();
            try {
                tailWaits.signal(); // try finally 快捷键： ctrl+alt+t
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }


    public static void main(String[] args) throws InterruptedException {
        BlockingQueue2<String> queue1 = new BlockingQueue2<>(3);
        /**Thread thread = new Thread(() -> {
            System.out.println(System.currentTimeMillis() + "begin");
            try {
                queue1.offer("任务1");
                System.out.println(queue1);
                queue1.offer("任务2");
                System.out.println(queue1);
                queue1.offer("任务3");
                System.out.println(queue1);
                queue1.offer("任务4", 5000);
                System.out.println(queue1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println(System.currentTimeMillis() + "end");
        }, "生产者1");
        thread.start();

        Thread.sleep(2000);
        queue1.poll(); **/
        new Thread(() -> {
            try {
                queue1.offer("任务1");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "offer").start();

        new Thread(() -> {
            try {
                queue1.poll();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "poll").start();
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("[");
        for (int i = 0; i < array.length; i++) {
            sb.append(array[i]);
            if (i != array.length - 1) {
                sb.append(",");
            }
        }
        sb.append("]");
        return sb.toString();
    }
}
